package event

import (
	"context"
	"fmt"
	"gitee.com/xwengf/go-event/v2/kafka"
	"gitee.com/xwengf/go-event/v2/rabbitmq"
	"gitee.com/xwengf/go-event/v2/redis"
	"time"

	"gitee.com/xwengf/go-event/v2/event"
)

type Receiver struct {
	receiver event.Receiver
	handler  event.Handler
}

type Server struct {
	receivers []Receiver
}

func NewServer() *Server {
	return &Server{
		receivers: make([]Receiver, 0),
	}
}

func (s *Server) Subscribe(r event.Receiver, h event.Handler) {
	s.receivers = append(s.receivers, Receiver{
		receiver: r,
		handler:  h,
	})
}

func (s *Server) Start(ctx context.Context) (err error) {
	fmt.Println("[Event] 启动事件服务")
	for _, r := range s.receivers {
		go r.receiver.Receive(ctx, r.handler)
	}
	return nil
}

func (s *Server) Stop(ctx context.Context) (err error) {
	fmt.Println("[Event] 关闭事件服务")
	return
}

type Options struct {
	Driver       string
	Brokers      []string
	QueueName    string
	User         string
	Password     string
	ReadTimeout  time.Duration
	WriteTimeout time.Duration
	ExchangeName string
}

func NewSender(c *Options) (event.Sender, error) {
	driver := c.Driver
	switch driver {
	case "rabbitmq":
		return rabbitmq.NewRabbitSender(c.Brokers[0], c.ExchangeName, c.QueueName)
	case "kafka":
		return kafka.NewKafkaSender(c.Brokers, c.User, c.Password)
	case "redis":
		s := redis.NewRedisSender(c.Brokers[0], c.User, c.Password, c.WriteTimeout, c.ReadTimeout)
		return s, nil
	}
	return nil, fmt.Errorf("driver %s not support", driver)
}

func NewReceiver(c *Options, topic string) (event.Receiver, error) {
	driver := c.Driver
	switch driver {
	case "rabbitmq":
		return rabbitmq.NewRabbitReceiver(c.Brokers[0], c.ExchangeName, c.QueueName, topic)
	case "kafka":
		return kafka.NewKafkaReceiver(c.Brokers, c.User, c.Password, []string{
			topic,
		}, c.QueueName)
	case "redis":
		j := redis.NewRedisReceiver(
			c.Brokers[0], //d.Redis.Addr,
			c.User, c.Password,
			topic,
			c.WriteTimeout,
			c.ReadTimeout)
		return j, nil
	}
	return nil, fmt.Errorf("driver %s not support", driver)
}
